package com.lifq.tmp;

import com.ververica.cdc.connectors.tidb.table.utils.TableKeyRangeUtils;
import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
import org.apache.flink.util.Preconditions;
import org.tikv.common.key.RowKey;
import org.tikv.common.util.KeyRangeUtils;
import org.tikv.kvproto.Coprocessor;
import org.tikv.shade.com.google.protobuf.ByteString;

import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;


/**
 * @Author: LiFuqiang
 * @Date: 2023/7/26 16:02
 * @Version 1.0
 * @description
 */
public class Tidb2DorisByCDC {

    public static void main(String[] args) throws Exception {
        long max = Long.MAX_VALUE;
        BigInteger aa = new BigInteger(longToBytes(max));
        System.out.println(aa.toString());

        long cv = -1L;
        byte[] cbv = longToBytes(cv);
        aa= new BigInteger("-1");
        System.out.println(aa.toString());
        byte[] cc2=  aa.toByteArray();

        BigInteger ddv = new BigInteger("18446744073709551620");
        byte[] ddb = ddv.toByteArray();
        System.out.println(ddb.length);


        System.out.println("pause");
        Coprocessor.KeyRange keyRange = getTableKeyRange(471, 3, 2);
        List<Coprocessor.KeyRange> keyRangeList = new ArrayList<>();
        keyRangeList.add(keyRange);

        List<Coprocessor.KeyRange> result = KeyRangeUtils.mergeSortedRanges(keyRangeList, 5);

        RowKey a1 = RowKey.toRowKey(471L,2L);
        RowKey a2 = RowKey.toRowKey(471L,3L);
        RowKey dd = RowKey.decode(keyRange.getEnd().toByteArray());

        RowKey b1 = RowKey.toRowKey(471L, 3074457345618258600L);
        RowKey b2 = RowKey.toRowKey(471L, 9223372036854775806L);
        RowKey bb = RowKey.decode(b2.getBytes());
        System.out.println("aa: " + Long.MAX_VALUE);
        System.out.println(Long.MAX_VALUE - 9223372036854775807L);
        RowKey rowKey = RowKey.decode(keyRange.getStart().toByteArray());

        keyCompare(keyRange.getStart(), keyRange.getEnd());
        keyCompare(keyRange.getStart(), keyRange.getStart());
        keyCompare(keyRange.getEnd(), keyRange.getStart());
        System.out.println(rowKey);


    }

    public static Coprocessor.KeyRange keyRangeAdd(ByteString start, long delta) {
        RowKey startRowKey = RowKey.decode(start.toByteArray());
        long aa = 0L;
        if (Long.MAX_VALUE - startRowKey.getHandle() <= delta) {
            aa = Long.MAX_VALUE;
        } else {
            aa = startRowKey.getHandle() + delta;
        }

        RowKey endRowKey = RowKey.toRowKey(startRowKey.getTableId(), aa);
        return KeyRangeUtils.makeCoprocRange(startRowKey.toByteString(), endRowKey.toByteString());
    }

    public static int keyCompare(ByteString start, ByteString end) {
        RowKey startRowKey = RowKey.decode(start.toByteArray());
        RowKey endRowKey = RowKey.decode(end.toByteArray());
        System.out.println(startRowKey.getHandle());
        System.out.println(endRowKey.getHandle());
        int aa = startRowKey.compareTo(endRowKey);
        if (aa > 0) {
            System.out.println("start 大于 end");
        } else if (aa == 0) {
            System.out.println("start 等于 end");
        } else {
            System.out.println("start 小于 end");
        }
        return aa;
    }

    public static Coprocessor.KeyRange getTableKeyRange(final long tableId) {
        return KeyRangeUtils.makeCoprocRange(
                RowKey.createMin(tableId).toByteString(),
                RowKey.createBeyondMax(tableId).toByteString());
    }

    public static List<Coprocessor.KeyRange> getTableKeyRanges(final long tableId, final int num) {
        Preconditions.checkArgument(num > 0, "Illegal value of num");

        if (num == 1) {
            return ImmutableList.of(getTableKeyRange(tableId));
        }

        final long delta =
                BigInteger.valueOf(Long.MAX_VALUE)
                        .subtract(BigInteger.valueOf(Long.MIN_VALUE + 1))
                        .divide(BigInteger.valueOf(num))
                        .longValueExact();
        final ImmutableList.Builder<Coprocessor.KeyRange> builder = ImmutableList.builder();
        for (int i = 0; i < num; i++) {
            final RowKey startKey =
                    (i == 0)
                            ? RowKey.createMin(tableId)
                            : RowKey.toRowKey(tableId, Long.MIN_VALUE + delta * i);
            final RowKey endKey =
                    (i == num - 1)
                            ? RowKey.createBeyondMax(tableId)
                            : RowKey.toRowKey(tableId, Long.MIN_VALUE + delta * (i + 1));
            System.out.println(startKey.getHandle());
            System.out.println(endKey.getHandle());
            builder.add(
                    KeyRangeUtils.makeCoprocRange(startKey.toByteString(), endKey.toByteString()));
        }
        return builder.build();
    }

    public static Coprocessor.KeyRange getTableKeyRange(final long tableId, final int num, final int idx) {
        Preconditions.checkArgument(idx >= 0 && idx < num, "Illegal value of idx");
        return getTableKeyRanges(tableId, num).get(idx);
    }

    public static boolean isRecordKey(final byte[] key) {
        return key[9] == '_' && key[10] == 'r';
    }

    public static byte[] longToBytes(long x) {
        ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
        buffer.putLong(x);
        return buffer.array();
    }

    public static long bytesToLong(byte[] bytes) {
        ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
        buffer.put(bytes);
        buffer.flip();//need flip
        return buffer.getLong();
    }
}
